package com.flink.chineseFilter;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Service
public class FlinkChineseFilterService {

    public String filterChinese(String input) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> inputStream = env.fromElements(input);

        DataStream<String> chineseStream = inputStream.flatMap((String value, org.apache.flink.util.Collector<String> out) -> {
            StringBuilder chinese = new StringBuilder();
            for (char c : value.toCharArray()) {
                if (Character.UnicodeScript.of(c) == Character.UnicodeScript.HAN) {
                    chinese.append(c);
                }
            }
            if (chinese.length() > 0) {
                out.collect(chinese.toString());
            }
        }).returns(String.class);

        List<String> result = new ArrayList<>();
        try (org.apache.flink.util.CloseableIterator<String> iterator = chineseStream.executeAndCollect()) {
            while (iterator.hasNext()) {
                result.add(iterator.next());
            }
        } catch (IOException e) {
            throw new RuntimeException("Error collecting Flink results", e);
        }

        if (!result.isEmpty()) {
            return result.get(0);
        }
        return "";
    }
}
